﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using A2DFramework.XmlService;
using A2DFramework.UniqueIdGeneratorService;

namespace A2DFramework.QueueService.Providers
{
    class RedisProvider : IQueueProvider
    {
        /// <summary>
        /// 永远返回true
        /// </summary>
        /// <param name="queueName"></param>
        /// <returns></returns>
        public bool IsQueueExists(string queueName)
        {
            using (var redis = RedisFactory.GetRedisClient())
            {
                return redis.ContainsKey(queueName);
            }
        }

        /// <summary>
        /// 建立队列
        /// </summary>
        /// <param name="queueName"></param>
        public void CreateQueue(string queueName)
        {
            //不作任何事情
            //Redis会自动建立队列
        }

        public void DeleteQueue(string queueName)
        {
            using (var redis = RedisFactory.GetRedisClient())
            {
                redis.Remove(queueName);
            }
        }

        public void SendMessage(string queueName, InnerMessageWrapper msg)
        {
            string listId = string.Format(QueueConfiguration.QueueConfiguration.RedisSettings.QueueNameFormat, queueName);
            using (var redis = RedisFactory.GetRedisClient())
            {
                msg.MessageID = UniqueIdGenerator.Next().ToString();

                
                byte[] bytes = SerializeUtils.Serialize2Bytes_Protobuf<InnerMessageWrapper>(msg);
                redis.LPush(listId, bytes);
            }
        }

        public InnerMessageWrapper ReceiveMessage(string queueName, string relatedMsgId = "")
        {
            InnerMessageWrapper obj = null;

            string listId = string.Format(QueueConfiguration.QueueConfiguration.RedisSettings.QueueNameFormat, queueName);
            string lockId = string.Format("{0}.lock", listId);

            using (var redis = RedisFactory.GetRedisClient())
            {
                if (string.IsNullOrEmpty(relatedMsgId))
                {
                    byte[] bytes = redis.BRPopValue(listId, 0);
                    if (bytes != null)
                        obj = SerializeUtils.DeserializeFromBytes_Protobuf<InnerMessageWrapper>(bytes);
                }
                else
                {
                    while (true)
                    {
                        byte[] bytes = redis.BRPopValue(listId, 0);

                        if (bytes == null)
                            continue;

                        if (bytes != null)
                            obj = SerializeUtils.DeserializeFromBytes_Protobuf<InnerMessageWrapper>(bytes);

                        if (obj == null)
                            continue;

                        if (obj.CorrelationID != relatedMsgId)
                            continue;

                        redis.LPush(queueName, bytes);
                        break;
                    }
                }
            }

            return obj;
        }
    }
}